/*
 * The MIT License (MIT)
 *
 * Copyright (c) 2016 pjoc.pub, blademainer.
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 *
 */

package pub.pjoc.pay.threadpool.asynchronous.impl;

import pub.pjoc.pay.threadpool.asynchronous.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.*;

/**
 * Created by xiongyingqi on 16-6-6.
 */
public class AsynchronousThreadMonitorImpl implements AsynchronousThreadMonitor, Runnable {
  private static final Logger logger = LoggerFactory
      .getLogger(AsynchronousThreadMonitorImpl.class);

  private ConcurrentHashMap<Class<?>, MonitorUnit> monitorMap = new ConcurrentHashMap<Class<?>, MonitorUnit>();

  private Thread thread;
  private AsynchronousInvokerService asynchronousInvoker;
  private boolean monitor = true;
  private long timeout = 0;
  private TimeUnit timeoutTimeUnit = TimeUnit.SECONDS;

  public AsynchronousThreadMonitorImpl(
      AsynchronousInvokerService asynchronousInvoker) {
    this.asynchronousInvoker = asynchronousInvoker;
  }

  @Override
  public void run() {
    while (monitor && !thread.isInterrupted()) {
      monitor();
    }
  }

  private void monitor() {
    BlockingQueue<FutureMonitor<Boolean>> futures = asynchronousInvoker.futureBlockingQueue();
    FutureMonitor<Boolean> take = null;
    try {
      take = futures.take();
    } catch (InterruptedException e) {
      logger.error("Take future monitor error with message: " + e.getMessage(), e);
    }
    if (logger.isDebugEnabled()) {
      logger.debug("Took a monitor: {}", take);
    }
    Future<Boolean> future = take.getFuture();
    Callable<Boolean> callable = take.getCallable();

    MonitorUnit monitorUnit = getOrInitMonitorUnit(callable);
    Timer timer = Timer.newTimer();
    try {

      Boolean completed;
      if (timeout > 0) {
        completed = future.get(timeout, timeoutTimeUnit);
      } else {
        completed = future.get();
      }

      if (completed == null) {
        monitorUnit.timeout(timer.time());
      } else if (!completed) {
        monitorUnit.fail(timer.time());
      } else {
        monitorUnit.success(timer.time());
      }

    } catch (InterruptedException e) {
      logger.error(
          "Caught exception when get invoke result with message: " + e.getMessage(),
          e);
    } catch (ExecutionException e) {
      logger.error(
          "Caught exception when get invoke result with message: " + e.getMessage(),
          e);
      monitorUnit.fail(timer.time());
    } catch (TimeoutException e) {
      monitorUnit.timeout(timer.time());
    }
  }

  private MonitorUnit getOrInitMonitorUnit(Callable<Boolean> callable) {
    MonitorUnit monitorUnit = monitorMap.get(callable.getClass());
    if (monitorUnit == null) {
      synchronized (callable.getClass()) {
        if (monitorMap.get(callable.getClass()) == null) {
          monitorUnit = new MonitorUnit(callable.getClass());
          monitorMap.put(callable.getClass(), monitorUnit);
        } else {
          monitorUnit = monitorMap.get(callable.getClass());
        }
      }
    }
    return monitorUnit;
  }

  @Override
  public void startMonitor() {
    synchronized (this) {
      thread = new Thread(this);
      thread.start();
    }
  }

  @Override
  public void stopMonitor() {
    synchronized (this) {
      monitor = false;
      thread.interrupt();
    }
  }

  @Override
  public int waitThreadSize() {
    ThreadPoolExecutor threadPool = asynchronousInvoker.getThreadPool();
    if (threadPool == null) {
      return 0;
    }
    BlockingQueue<Runnable> queue = threadPool.getQueue();
    if (queue == null) {
      return 0;
    }
    int size = queue.size();
    if (logger.isDebugEnabled()) {
      logger.debug("Getting wait thread size with: {}", size);
    }
    return asynchronousInvoker.getThreadPool().getQueue().size();
  }

  @Override
  public long averageExecuteTime() {
    Collection<MonitorUnit> values = monitorMap.values();
    if (values.isEmpty()) {
      return 0;
    }
    BigDecimal countSum = new BigDecimal(0);
    BigDecimal averageSum = new BigDecimal(0);
    for (MonitorUnit value : values) {
      BigDecimal count = value.getCount();
      countSum = countSum.add(count);
      BigDecimal averageTime = value.getAverageTime();
      averageSum = averageSum.add(averageTime);
    }
    if (countSum.longValue() == 0) {
      return 0;
    }
    BigDecimal divide = MonitorUnit.divide(averageSum, countSum);
    return divide.longValue();
  }

  @Override
  public long timeOutSize() {
    Collection<MonitorUnit> values = monitorMap.values();
    if (values.isEmpty()) {
      return 0;
    }
    BigDecimal countSum = new BigDecimal(0);
    for (MonitorUnit value : values) {
      BigDecimal count = value.getTimeoutCount();
      countSum = countSum.add(count);
    }
    return countSum.longValue();
  }

  @Override
  public Map<Class<?>, MonitorUnit> getMonitorMap() {
    return monitorMap;
  }

}
